package com.lg.service;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutionException;

@Slf4j
@Service
public class KafkaService {

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	public String sendMessage(ProducerRecord<String, String> record) throws ExecutionException, InterruptedException {

		SendResult<String, String> result = this.kafkaTemplate.send(record).get();
		RecordMetadata metadata = result.getRecordMetadata();
		String returnResult = metadata.topic() + "\t" + metadata.partition() + "\t" + metadata.offset();
		log.info("发送消息成功：{}", returnResult);

		return returnResult;
	}

}